package org.nbict.iot.task.once;

import com.alibaba.fastjson.JSONObject;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.MapFunction;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.nbict.iot.task.once.charge.ChargeStatus;
import org.nbict.iot.task.once.journey.JourneyStatus;
import org.nbict.iot.task.once.state.RunningWinStateFactory;
import org.nbict.iot.task.once.state.RunningWinStateUpdater;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;

/**
 * 实时计算车辆旅程信息
 * Created by songseven on 18/7/10.
 */
public class RuningWinTrident {

    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        KafkaSpoutConfig.FirstPollOffsetStrategy strategy = KafkaSpoutConfig
                .FirstPollOffsetStrategy.EARLIEST;
        if (args.length > 1) {
            strategy = KafkaSpoutConfig.FirstPollOffsetStrategy.valueOf
                    (args[1]); //strategy name
        }

        TridentTopology topology = new TridentTopology();
        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server,192.168.1.83:9092 transit-.*-simple
                        "iot-car-group-window", strategy); // group id ignored
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-parsed-4-window", kafkaSpout);

        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");
        Stream stage = stream.each(new Fields("platform_id", "packet"), new
                PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header",
                        "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action",
                        "timestamp", "upload"));
//
//        Stream uploadStream = stage.each(outputFields, new UploadFilter())
//                .partitionBy(new Fields("car_vin"))
//                .toStream(); //2 or 3

        stage.partitionBy(new Fields("car_vin")).map(
                //构造点
                new MapFunction() {

                    private int count = 0;
                    @Override
                    public Values execute(TridentTuple tuple) {
                        //platform_id --0, car_vin --1,  action --2,  timestamp --3, upload --4
                        String vin = tuple.getString(1);
                        Long timestamp = tuple.getLong(3);
                        JSONObject upload = (JSONObject) tuple.get(4); //info
                        Integer action = tuple.getInteger(2);
                        Integer soc = null;
                        Long lon = null;
                        Long lat = null;

                        JourneyStatus status = null;
                        ChargeStatus chargeStatus = ChargeStatus.UNCHARGED;
                        switch (action) {
                            case 1: //login
                                status = JourneyStatus.LOGIN;
                                break;
                            case 2://realtime
                            case 3://replay
                                JSONObject info = (JSONObject) upload.get("01");
                                JSONObject gis = (JSONObject)upload.get("05");
                                if (info == null || gis == null) break;
                                String s = info.getString("status");
                                soc = info.getInteger("soc");
                                lon = gis.getLong("lon");
                                lat = gis.getLong("lat");
                                if (s.equals("01")) {
                                    status = JourneyStatus.START;
                                } else if (s.equals("02")) {
                                    status = JourneyStatus.END;
                                }
                                String charged = info.getString("charged_state");
                                if (charged.equals("01")) {
                                    chargeStatus = ChargeStatus.CHARGING;//停车充电
                                } else if (charged.equals("04")) {
                                    chargeStatus = ChargeStatus.DONE; //充电完成
                                }
                                break;
                            case 4://logout
                                status = JourneyStatus.LOGOUT;
                                break;
                        }

                        //System.out.println(count++);
                        return new Values(new Pointer(vin, timestamp, status, soc, lon, lat, chargeStatus)); //经度和维度
                    }
                },new Fields("point")).project(new Fields("point"))
                .partitionPersist(new RunningWinStateFactory(),
                        new Fields("point"),
                        new RunningWinStateUpdater(),
                        new Fields()); //output

        Config config = new Config();
        config.setMaxSpoutPending(3);
        //config.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 60); //定时


        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-gis", config, topology.build());
        }

    }

}
